package com.atguigu.flink.chapter7;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/22 11:36
 */
public class Flink05_Chapter7_Window_Function {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
          .socketTextStream("hadoop162", 9999)
          .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
              @Override
              public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                  String[] s = value.split(" ");
                  for (String word : s) {
                      out.collect(Tuple2.of(word, 1L));
                  }
              }
          })

          .keyBy(t -> t.f0)
          .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
          /*.reduce(new ReduceFunction<Tuple2<String, Long>>() {
              @Override
              public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                  System.out.println("reduce...");
                  return Tuple2.of(value1.f0, value1.f1 + value2.f1);
              }
          })*/
          /*.aggregate(new AggregateFunction<Tuple2<String, Long>, Long, String>() {
              // 创建累加器
              @Override
              public Long createAccumulator() {
                  System.out.println("createAccumulator...");
                  return 0L;
              }
              // 聚合
              @Override
              public Long add(Tuple2<String, Long> value, Long accumulator) {
                  System.out.println("add...");
                  return accumulator + value.f1;
              }
              // 返回窗口关闭之后的聚合结果
              @Override
              public String getResult(Long accumulator) {
                  System.out.println("getResult...");
                  return "最终结果: " + accumulator;
              }
              // 合并累计器  只有在 SessionWindow才会使用
              @Override
              public Long merge(Long a, Long b) {
                  System.out.println("merge...");
                  return a + b;
              }
          })*/
          .process(new ProcessWindowFunction<Tuple2<String, Long>, Long, String, TimeWindow>() {
              @Override
              public void process(String key,
                                  Context context,
                                  Iterable<Tuple2<String, Long>> elements,
                                  Collector<Long> out) throws Exception {
                  System.out.println("process...");
              }
          })
          .print();

        env.execute();
    }
}
/*
reduce
    增量聚合
        来一条做一次聚合
        当窗口关闭的时候的返回结果

        返回值的类型要和输入类型保持一致
aggregate
    增量聚合
        来一条做一次聚合
        当窗口关闭的时候的返回结果

        返回值的类型要和输入类型不要钱必须一致

process
    ProcessWindowFunction



 */